package org.fujene.communicate.deleter;

import java.util.Map.Entry;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.concurrent.LinkedBlockingQueue;
import org.fujene.communicate.MainServiceSettings;
import org.fujene.communicate.hash.Hashing;
import org.fujene.communicate.node.Node;
import org.fujene.communicate.rpc.Communicator;

public class DeleterHost implements Runnable{
	private static SortedSet<Node> nodes;
	private static Node myNode;
	private static int replication;
	private static LinkedBlockingQueue<DeleterPackage> deleteQueue;
	private Communicator communicator;
	private int deleteKeyType;

	static{
		deleteQueue = new LinkedBlockingQueue<DeleterPackage>();
	}

	public DeleterHost(){
		communicator = new Communicator();
	}

	public static void addToQueue(DeleterPackage pkg){
		deleteQueue.add(pkg);
	}

	public static boolean isEmpty(){
		return deleteQueue.isEmpty();
	}

	public static int residue(){
		return deleteQueue.size();
	}

	public static void property(int repl, Node ownNode, SortedSet<Node> list){
		replication = repl;
		myNode = ownNode;
		nodes = list;
	}

	@Override
	public void run(){
		DeleterPackage pkg = null;
		while(true){
			try{
				pkg = deleteQueue.take();
			}catch(InterruptedException e1){
				continue;
			}

			if((deleteKeyType == MainServiceSettings.DELTYPE_CONTENT || deleteKeyType == MainServiceSettings.DELTYPE_APPENDIX)
				&& pkg.type() != DeleterPackage.TYPE_CONTENT){
				continue;
			}
			if((deleteKeyType == MainServiceSettings.DELTYPE_ID || deleteKeyType == MainServiceSettings.DELTYPE_TIME)
				&& pkg.type() != DeleterPackage.TYPE_ID){
				continue;
			}
			if((deleteKeyType == MainServiceSettings.DELTYPE_MULTIFLAG)
				&& pkg.type() != DeleterPackage.TYPE_FLAG){
				continue;
			}
			long hash = Hashing.getContentHash(pkg.ID());
			TreeMap<Long, Node> hashList = new TreeMap<Long, Node>();
			Entry<Long, Node> currentEntry = null;
			for(Node eachNode: nodes)
				hashList.put(eachNode.getAttributeHash(hash), eachNode);
			currentEntry = hashList.ceilingEntry(Long.valueOf(hash));
			if(currentEntry == null) currentEntry = hashList.firstEntry();
			Node nearest = currentEntry.getValue();
			int cluster = nodes.headSet(nearest).size();
			pkg.setCluster(cluster);

			for(int i = 0; i < (replication > nodes.size()? nodes.size(): replication); i++){
				if(currentEntry == null) currentEntry = hashList.ceilingEntry(Long.valueOf(hash));
				else currentEntry = hashList.higherEntry(currentEntry.getKey());
				if(currentEntry == null) currentEntry = hashList.firstEntry();

				if(currentEntry.getValue() != myNode){
					// remote
					communicator.deleteContent(currentEntry.getValue(), pkg);
				}else{
					// local
					DeleterClient.addToQueue(pkg);
				}
			}
		}
	}
}
